package com.mpsc.spark.demo.controller;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.web.bind.annotation.RestController;
import scala.Tuple2;

import javax.annotation.Resource;
import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;

@RestController
public class SparkController {
    @Resource
    private SparkSession sparkSession;

    @Resource
    private JavaSparkContext javaSparkContext;

    public void testSparkText() {
        String file = "D:\\TEMP\\word.txt";
        JavaRDD<String> fileRDD =  javaSparkContext.textFile(file);

        JavaRDD<String> wordsRDD = fileRDD.flatMap(line -> Arrays.asList(line.split(" ")).iterator());
        JavaPairRDD<String, Integer> wordAndOneRDD = wordsRDD.mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> wordAndCountRDD = wordAndOneRDD.reduceByKey((a, b) -> a + b);

        //输出结果
        List<Tuple2<String, Integer>> result = wordAndCountRDD.collect();
        result.forEach(System.out::println);
    }

    public void testSparkCsv() {
        String file = "D:\\TEMP\\testcsv.csv";
        JavaRDD<String> fileRDD = javaSparkContext.textFile(file);
        JavaRDD<String> wordsRDD = fileRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator());

        //输出结果
        System.out.println(wordsRDD.collect());
    }

    public void testSparkMysql() throws IOException {

        Dataset<Row> jdbcDF = sparkSession.read()
                .format("jdbc")
                .option("url", "jdbc:mysql://192.168.140.1:3306/user?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai")
                .option("dbtable", "(SELECT * FROM xxxtable) tmp")
                .option("user", "root")
                .option("password", "xxxxxxxxxx*k")
                .option("driver", "com.mysql.cj.jdbc.Driver")
                .load();

        jdbcDF.printSchema();
        jdbcDF.show();

        //转化为RDD
        JavaRDD<Row> rowJavaRDD = jdbcDF.javaRDD();
        System.out.println(rowJavaRDD.collect());

        List<Row> list = rowJavaRDD.collect();
        BufferedWriter bw;
        bw = new BufferedWriter(new FileWriter("d:/test.txt"));
        for (int j = 0; j < list.size(); j++) {
            bw.write(list.get(j).toString());
            bw.newLine();
            bw.flush();
        }
        bw.close();
    }

    public void testSparkJson() {
        Dataset<Row> df = sparkSession.read().json("D:\\TEMP\\testjson.json");
        df.printSchema();

        df.createOrReplaceTempView("t");
        Dataset<Row> row = sparkSession.sql("select age,name from t where age > 3");

        JavaRDD<Row> rowJavaRDD = row.javaRDD();
        System.out.println(rowJavaRDD.collect());
    }

    public void testSparkCsv1() {
        String file = "D:\\TEMP\\testcsv.csv";
        JavaRDD<String> fileRDD = javaSparkContext.textFile(file);
        JavaRDD<String> wordsRDD = fileRDD.flatMap(line -> Arrays.asList(line.split(",")).iterator());

        //输出结果
        System.out.println(wordsRDD.collect());
    }
}
